package com.zsy.filink.basic;

import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @Author: Zhou ShiYang
 * @Description: 第一个基于Flink实时处理快速入门案例
 * @Date:Create：in 2021/7/27 23:29
 **/
public class StreamingWCApp {

    public static void main(String[] args) throws Exception{

        // 创建上下文
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 对接数据源数据，数据格式：pk,pk,pk,flink,flink
        DataStreamSource<String> source = env.socketTextStream("localhost", 9527);

        // 业务处理逻辑  transformation
        source.flatMap(new FlatMapFunction<String, String>() {
            // 字符串按逗号切割，全部小写，去除首尾空格，Collector输出
            @Override
            public void flatMap(String value, Collector<String> out) throws Exception {
                String[] words = value.split(",");
                for (String word : words) {
                    out.collect(word.toLowerCase().trim());
                }
            }
        }).filter(new FilterFunction<String>() {
            // 过滤掉空字符串
            @Override
            public boolean filter(String value) throws Exception {
                return StringUtils.isNotEmpty(value);
            }
        }).map(new MapFunction<String, Tuple2<String, Integer>>() {
            // 映射为，key：word(单词)，value：1(次数)
            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {
                return new Tuple2<>(value, 1);
            }
        }).keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
            // 根据上面map出来的key作为标识进行统计（累加value）
            @Override
            public String getKey(Tuple2<String, Integer> value) throws Exception {
                // value.f0：第0位
                return value.f0;
            }
            // 第1位求和
        }).sum(1).print();

        // 执行
        env.execute("StreamingWCApp");

    }
}
